package com.hilin.hilinim.netty.ws;

import com.hilin.entity.vo.RespVo;
import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.util.AttributeKey;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.ObjectUtils;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class WsUtils {

    /**
     * 用户orgId所在的key
     */
    private static final AttributeKey<Long> USER_ORGID_KEY = AttributeKey.newInstance("im_user_orgid");
    /**
     * 用户accountId所在的key
     */
    private static final AttributeKey<String> USER_ACCOUNTID_KEY = AttributeKey.newInstance("im_user_accountid");

    /**
     * 用户最后发送消息时间--包括心跳
     */
    private static final AttributeKey<Long> USER_LAST_MSG_TIME_KEY = AttributeKey.newInstance("im_user_last_msg_time");

    /**
     * 用户认证结果
     */
    private static final AttributeKey<RespVo<Object>> USER_AUTH_RESULT_KEY = AttributeKey.newInstance("im_user_auth_result");


    /**
     * key 用户订阅的key
     * value 用户列表
     */
    private static final Map<String, ChannelGroup> SUB_USERS = new ConcurrentHashMap<>();

    /**
     * 所有用户
     */
    private static final String SK_ALL = "sub_im_all";
    /**
     * 根据orgId查询
     */
    private static final String SK_ORGID = "sub_im_orgid_%s";
    /**
     * 根据orgid和accountId查询
     */
    private static final String SK_ORGID_ACCOUNTID = "sub_im_orgid_accountid_%s_%s";

    /**
     * 心跳发送速率，单位:秒/次
     */
    private static long heartRate = 30;
    /**
     * 心跳超时时间，单位:秒
     */
    private static long heartTimeout = 90;

    public static void setHeartRate(long heartRate) {
        WsUtils.heartRate = heartRate;
    }

    public static void setHeartTimeout(long heartTimeout) {
        WsUtils.heartTimeout = heartTimeout;
    }

    static {
        SUB_USERS.putIfAbsent(SK_ALL, new DefaultChannelGroup(SK_ALL, GlobalEventExecutor.INSTANCE));
        // 心跳线程
        new Thread(() -> {
            while (true) {
                try {
                    Thread.sleep(heartRate * 1000);
                    send(new PingWebSocketFrame());
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();

        // 心跳检测线程
        new Thread(() -> {
            while (true) {
                long currTime = System.currentTimeMillis();
                SUB_USERS.get(SK_ALL).forEach(channel -> {
                    if (ObjectUtils.isEmpty(getMsgLastTime(channel))) {
                        setMsgLastTime(channel, System.currentTimeMillis());
                    } else {
                        long lastTime = getMsgLastTime(channel);
                        if (lastTime + heartTimeout * 1000 < currTime) {
                            // 心跳超时
                            channel.close();
                        }
                    }
                });
            }
        }).start();
    }

    /**
     * 添加用户
     */
    public static void add(Channel channel, Long orgId, String accountId) {
        setOrgId(channel, orgId);
        setAccountId(channel, accountId);

        String subOrgId = String.format(SK_ORGID, orgId);
        SUB_USERS.putIfAbsent(subOrgId, new DefaultChannelGroup(subOrgId, GlobalEventExecutor.INSTANCE));
        SUB_USERS.get(subOrgId).add(channel);

        String subOrgIdAccountId = String.format(SK_ORGID_ACCOUNTID, orgId, accountId);
        SUB_USERS.putIfAbsent(subOrgIdAccountId, new DefaultChannelGroup(subOrgIdAccountId, GlobalEventExecutor.INSTANCE));
        SUB_USERS.get(subOrgIdAccountId).add(channel);

        SUB_USERS.get(SK_ALL).add(channel);

        log.info("用户加入，在线人数:{}", getUserCount());
    }

    /**
     * 移除用户
     */
    public static void remove(Channel channel) {
        if (!checkChannel(channel)) {
            return;
        }
        remove(getOrgId(channel), getAccountId(channel), channel);
        log.info("用户退出，在线人数:{}", getUserCount());
    }

    private static void remove(String subKey, Channel channel) {
        ChannelGroup group = SUB_USERS.get(subKey);
        if (!ObjectUtils.isEmpty(group)) {
            group.remove(channel);
        }
        if (!SK_ALL.equals(subKey)) {
            if (ObjectUtils.isEmpty(group)) {
                SUB_USERS.remove(subKey);
            }
        }
    }

    private static void remove(Long orgId, String accountId, Channel channel) {
        remove(SK_ALL, channel);
        remove(String.format(SK_ORGID, orgId), channel);
        remove(String.format(SK_ORGID_ACCOUNTID, orgId, accountId), channel);
    }

    /**
     * 发送到所有用户
     *
     * @param msg
     */
    public static void send(WebSocketFrame msg) {
        send(SUB_USERS.get(SK_ALL), msg);
    }

    /**
     * 发送到某个平台所有用户
     *
     * @param orgId
     * @param msg
     */
    public static void send(Long orgId, WebSocketFrame msg) {
        send(SUB_USERS.get(String.format(SK_ORGID, orgId)), msg);
    }

    /**
     * 发送到某个用户
     *
     * @param accountId
     * @param orgId
     * @param msg
     */
    public static void send(Long orgId, String accountId, WebSocketFrame msg) {
        send(SUB_USERS.get(String.format(SK_ORGID_ACCOUNTID, orgId, accountId)), msg);
    }

    public static void send(ChannelGroup group, WebSocketFrame frame) {
        if (!ObjectUtils.isEmpty(group)) {
            group.writeAndFlush(frame);
        }
    }

    public static long getUserCount() {
        return SUB_USERS.get(SK_ALL).size();
    }

    public static boolean checkChannel(Channel channel) {
        return channel != null
                && channel.hasAttr(USER_ORGID_KEY)
                && !ObjectUtils.isEmpty(channel.attr(USER_ORGID_KEY).get())
                && channel.hasAttr(USER_ACCOUNTID_KEY)
                && !ObjectUtils.isEmpty(channel.attr(USER_ACCOUNTID_KEY).get());
    }

    /**
     * 获取当前通道的用户ID
     *
     * @param channel
     * @return
     */
    public static String getAccountId(Channel channel) {
        return channel.attr(USER_ACCOUNTID_KEY).get();
    }

    /**
     * 获取当前通道的appid
     *
     * @param channel
     * @return
     */
    public static Long getOrgId(Channel channel) {
        return channel.attr(USER_ORGID_KEY).get();
    }

    /**
     * 设置当前通道的最后收到消息时间
     *
     * @param channel
     * @return
     */
    public static Long getMsgLastTime(Channel channel) {
        return channel.attr(USER_LAST_MSG_TIME_KEY).get();
    }

    /**
     * 获取当前通道认证结果
     *
     * @param channel
     * @return
     */
    public static RespVo<Object> getUserAuthResult(Channel channel) {
        return channel.attr(USER_AUTH_RESULT_KEY).get();
    }

    /**
     * 设置当前通道的accountId
     *
     * @param channel
     * @return
     */
    public static void setAccountId(Channel channel, String accountId) {
        channel.attr(USER_ACCOUNTID_KEY).set(accountId);
    }

    /**
     * 设置当前通道的orgId
     *
     * @param channel
     * @return
     */
    public static void setOrgId(Channel channel, Long orgId) {
        channel.attr(USER_ORGID_KEY).set(orgId);
    }

    /**
     * 设置当前通道的最后收到消息时间
     *
     * @param channel
     * @return
     */
    public static void setMsgLastTime(Channel channel, long msgLastTime) {
        channel.attr(USER_LAST_MSG_TIME_KEY).set(msgLastTime);
    }

    /**
     * 设置当前通道认证结果
     *
     * @param channel
     * @return
     */
    public static void setUserAuthResult(Channel channel, RespVo<Object> authResult) {
        channel.attr(USER_AUTH_RESULT_KEY).set(authResult);
    }

}
